package com.lvmama.rhino.message;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;

import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
import kafka.utils.VerifiableProperties;

import org.apache.log4j.Logger;

/**
 * 
 * @author wuyunpeng
 * 
 * @param <T>
 */
public class ObjectEncoderAndDecoder<T> implements Encoder<T>, Decoder<T> {

	private static final Logger LOGGER = Logger.getLogger(ObjectEncoderAndDecoder.class);
	
	public ObjectEncoderAndDecoder(VerifiableProperties prop) {
		prop = null;
	}

	public byte[] toBytes(T obj) {

		try {
			ByteArrayOutputStream bo = new ByteArrayOutputStream();
			ObjectOutputStream oo = new ObjectOutputStream(bo);
			oo.writeObject(obj);
			return bo.toByteArray();
		} catch (Exception e) {
			e.printStackTrace();
			LOGGER.info("Failed to serialize object");
			return null;
		}
	}

	@SuppressWarnings("unchecked")
	public T fromBytes(byte[] bytes) {
		
		try {
			ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
			ObjectInputStream oi = new ObjectInputStream(bi);
			T obj = (T) oi.readObject();
			oi.close();
			bi.close();
			return obj;
		} catch (Exception e) {
			e.printStackTrace();
			LOGGER.info("Failed to deserizlize object");
			return null;
		}
	}
	
}
